package com.qingyunge.app.dwd;

import com.alibaba.fastjson.JSONAware;
import com.alibaba.fastjson.JSONObject;
import com.qingyunge.common.KafkaConfig;
import com.qingyunge.util.MyKafkaUtil;
import com.qingyunge.util.NumberExtractorUtil;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class Dwd_sku_produce_filter {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
        String topic = KafkaConfig.TOPIC_ID;
        String gourpid = "dwd_produce1";
        OutputTag<String> productEvalTag = new OutputTag<String>("product_eval"){};
        OutputTag<String> errorTag = new OutputTag<String>("error"){};
        SingleOutputStreamOperator<JSONObject> jsonObjectSingleOutputStreamOperator = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, gourpid))
                .process(new ProcessFunction<String, JSONObject>() {
                    @Override
                    public void processElement(String s, Context context, Collector<JSONObject> collector) throws Exception {
                        try {
                            JSONObject jsonObject = JSONObject.parseObject(s);
                            String table = jsonObject.getString("table");
                            String data12 = jsonObject.getJSONObject("data").getString("id");
                            if(table != null && data12 != null){
                                if (table.equals("product_eval")){
                                    JSONObject data = jsonObject.getJSONObject("data");
                                    String location = data.getString("location");
                                    if (location == null || location.equals("")){
                                        data.put("location","中国");
                                        jsonObject.put("data",data);
                                    }
                                    context.output(productEvalTag,jsonObject.toJSONString());
                                }
                                if(table.equals("sku_info")){
                                    JSONObject data = jsonObject.getJSONObject("data");
                                    String evalType = data.getString("eval_type");
                                    if (evalType == null || evalType.equals("") || evalType.equals("新品上市")){
                                        data.put("eval_num",0);
                                        jsonObject.put("data",data);
                                    }else {
                                        String s1 = NumberExtractorUtil.get(evalType);
                                        if(s1==null){
                                            data.put("eval_num",0);
                                            jsonObject.put("data",data);
                                        }else{
                                            if(evalType.contains("万")){
                                                data.put("eval_num",Long.parseLong(s1) * 10000L);
                                                jsonObject.put("data",data);
                                            }else{
                                                data.put("eval_num",Long.parseLong(s1));
                                                jsonObject.put("data",data);
                                            }
                                        }
                                    }
                                    String porView = data.getString("por_view");
                                    if(porView == null || porView.equals("")){
                                        data.put("por_view",90);
                                    }else {
                                        String s1 = NumberExtractorUtil.get(porView);
                                        if (s1==null){
                                            data.put("por_view",50);
                                        }else{
                                            data.put("por_view",Integer.valueOf(s1));
                                        }
                                    }
                                    jsonObject.put("data",data);
                                    collector.collect(jsonObject);
                                }
                            }else {
                                context.output(errorTag, jsonObject.toJSONString());
                            }
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                });
        DataStream<String> productEvalDS = jsonObjectSingleOutputStreamOperator.getSideOutput(productEvalTag);
        DataStream<String> errorDS = jsonObjectSingleOutputStreamOperator.getSideOutput(errorTag);
        SingleOutputStreamOperator<String> skuDS = jsonObjectSingleOutputStreamOperator.map(JSONAware::toJSONString);
        String productEval_topic = "dwd_traffic_eval";
        String sku_topic = "dwd_traffic_sku";
        String error_topic = "dwd_traffic_error";
        productEvalDS.print("product===");
        skuDS.print("sku===");
        errorDS.print("error=========");
        productEvalDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(productEval_topic));
        skuDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(sku_topic));
        env.execute("DwdSkuProduceFilter");
    }
}
